package com.example.wxg.example.stock;

import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @author void
 * @date 2022/1/5 15:39
 * @desc
 */
public class StockMain {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StockSource stockSource = new StockSource("D:\\ideaWorkSpace\\spring-boot-test\\flink\\src\\main\\resources\\stock\\file");
        
        DataStreamSource<StockPrice> dataStreamSource = env.addSource(stockSource);
        //价格最大值
        DataStream<StockPrice> dataStream1 = dataStreamSource.keyBy("symbol").max("price");
        //汇率折算
        DataStream<StockPrice> dataStream2 = dataStreamSource.map(s->{
            s.setPrice(s.getPrice()*7);
            return s;
        });
        //大额交易过滤
        DataStream<StockPrice> dataStream3 = dataStreamSource.filter(s->s.getVolume()>150);

        dataStream1.print();
        dataStream2.print();
        dataStream3.print();

        dataStreamSource.print();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        env.execute("开始股票分析任务");
    }
}
